package com.rn.gypsophila.kafka.producer;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Description: 生产者
 *
 * @author 然诺
 * @date 2021/01/29
 */
@Component
public class KafkaProducer {

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	/**
	 * 发送消息（简单）
	 */
	public void sendMessage(String channel, String message) {
		try {
			SendResult<String, String> sendResult = kafkaTemplate.send(channel, message).get(10, TimeUnit.SECONDS);
			RecordMetadata recordMetadata = sendResult.getRecordMetadata();
			System.out.println("发送消息成功:" + recordMetadata.topic() + "-" + recordMetadata.partition() + "-" + recordMetadata.offset());
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 发送消息（指定分区）
	 */
	public void sendMessageWithPartition(String channel, int partition, String message) {
		kafkaTemplate.send(channel, partition, "key_tt", message);
	}

	/**
	 * 发送消息（异步带回调-在回调方法中监控消息是否发送成功 或 失败时做补偿处理）
	 */
	public void sendMessageWithCallback(String channel, String message) {
		kafkaTemplate.send(channel, message).addCallback(sendResult -> {
			if (sendResult != null) {
				RecordMetadata recordMetadata = sendResult.getRecordMetadata();
				System.out.println("发送消息成功:" + recordMetadata.topic() + "-" + recordMetadata.partition() + "-" + recordMetadata.offset());
			}
		}, failure -> System.out.println("发送消息失败:" + failure.getMessage()));
	}

	/**
	 * 发送消息（异步带回调-写法2）
	 */
	public void sendMessageWithCallback2(String channel, String message) {
		kafkaTemplate.send(channel, message).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
			@Override
			public void onFailure(Throwable throwable) {
				System.out.println("发送消息失败:" + throwable.getMessage());
			}

			@Override
			public void onSuccess(SendResult<String, String> sendResult) {
				RecordMetadata recordMetadata = sendResult.getRecordMetadata();
				System.out.println("发送消息成功:" + recordMetadata.topic() + "-" + recordMetadata.partition() + "-" + recordMetadata.offset());
			}
		});
	}

}
